package com.atguigu.gmall.activity.recevier;

import com.atguigu.gmall.activity.model.SeckillGoods;
import com.atguigu.gmall.activity.model.UserRecode;
import com.atguigu.gmall.activity.service.SeckillGoodsService;
import com.atguigu.gmall.common.constant.MqConst;
import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.common.util.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.Date;
import java.util.List;

/**
 * @author: atguigu
 * @create: 2023-01-30 14:10
 */
@Slf4j
@Component
public class SeckillReceiver {

    @Autowired
    private SeckillGoodsService seckillGoodsService;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 缓存预热
     * 监听定时服务消息，将秒杀库中商品加入到缓存
     * 所有秒杀商品 hash 结构    redisKey seckill:goods   hashKey:skuId hashVal:秒杀商品表中商品信息
     * 某个秒杀商品库存 List结构  Key seckill:stock:+skuId  vals skuId
     *
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(MqConst.EXCHANGE_DIRECT_TASK),
            value = @Queue(MqConst.QUEUE_TASK_1),
            key = MqConst.ROUTING_TASK_1
    ))
    public void importSeckillGoodsToRedis(Message message, Channel channel) {
        try {
            //1.查询秒杀表中商品 条件：审核状态必须为“1”  秒杀商品库存大于0  秒杀开始时间是当日
            LambdaQueryWrapper<SeckillGoods> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(SeckillGoods::getStatus, "1");
            queryWrapper.gt(SeckillGoods::getStockCount, 0);
            queryWrapper.apply("DATE_FORMAT(start_time,'%Y-%m-%d') = '" + DateUtil.formatDate(new Date()) + "'");
            List<SeckillGoods> seckillGoodsList = seckillGoodsService.list(queryWrapper);

            //2.将秒杀商品遍历存入缓存
            if (!CollectionUtils.isEmpty(seckillGoodsList)) {
                String redisKey = RedisConst.SECKILL_GOODS;
                BoundHashOperations<String, String, SeckillGoods> hashOps = redisTemplate.boundHashOps(redisKey);
                for (SeckillGoods seckillGoods : seckillGoodsList) {
                    //2.1 判断缓存中已有秒杀商品
                    Boolean flag = hashOps.hasKey(seckillGoods.getSkuId().toString());
                    if (flag) {
                        continue;
                    }
                    //2.2 将秒杀商品存入Redis Hash结构存储所有秒杀商品
                    hashOps.put(seckillGoods.getSkuId().toString(), seckillGoods);
                    //2.3 为每个秒杀商品设置库存避免超卖  List结构

                    String stockKey = RedisConst.SECKILL_STOCK_PREFIX + seckillGoods.getSkuId();
                    BoundListOperations listOps = redisTemplate.boundListOps(stockKey);
                    for (Integer i = 0; i < seckillGoods.getStockCount(); i++) {
                        listOps.leftPush(seckillGoods.getSkuId().toString());
                    }
                    //2.4 商品被加入到分布式缓存后，需要通知每个秒杀服务更新本地缓存中 商品状态
                    redisTemplate.convertAndSend("seckillpush", seckillGoods.getSkuId() + ":1");
                }
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 从秒杀队列中获取获取秒杀请求，进行秒杀订单预处理
     *
     * @param userRecode
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(MqConst.EXCHANGE_DIRECT_SECKILL_USER),
            value = @Queue(MqConst.QUEUE_SECKILL_USER),
            key = MqConst.ROUTING_SECKILL_USER
    ))
    public void processSeckillRequest(UserRecode userRecode, Message message, Channel channel) {
        try {
            if (userRecode != null) {
                log.info("【秒杀服务】监听到秒杀请求：{}", userRecode);
                seckillGoodsService.processPreSeckillOrder(userRecode);
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("【秒杀服务】监听到秒杀请求异常：", e);
        }
    }


    /**
     * 监听秒杀扣减库存
     *
     * @param skuId   商品ID
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(MqConst.EXCHANGE_DIRECT_SECKILL_STOCK),
            value = @Queue(MqConst.QUEUE_SECKILL_STOCK),
            key = MqConst.ROUTING_SECKILL_STOCK
    ))
    public void updateStock(Long skuId, Message message, Channel channel) {
        try {
            if (skuId != null) {
                log.info("【秒杀服务】监听到秒杀扣减库存消息：{}", skuId);
                seckillGoodsService.updateStock(skuId);
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("【秒杀服务】监听到秒杀扣减异常：{}", e);
        }
    }


    /**
     * 清理秒杀产生相关缓存数据，更新数据库中秒杀商品状态
     */
    @RabbitListener(bindings =@QueueBinding(
            exchange = @Exchange(MqConst.EXCHANGE_DIRECT_TASK),
            value = @Queue(MqConst.QUEUE_TASK_18),
            key = MqConst.ROUTING_TASK_18
    ))
    public void closeSeckillOrderAndClearCache(Message message, Channel channel){
        try {
            log.info("【秒杀服务】监听到清理秒杀缓存消息");
            seckillGoodsService.closeSeckillOrderAndClearCache();
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
